package com.wandian.texas;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import com.wandian.texas.protocol.Protocol;
import com.wandian.util.Log;

/**
 * @author long
 * @date 2012-12-7 下午4:53:03
 * @qq 108232706
 * @description Requestor
 */
public abstract class Requestor extends Thread {
	public SocketChannel socketChannel = null;
	public Selector selector = null;
	public ReentrantLock selectorLock = new ReentrantLock();
	public boolean isService = false;
	public boolean isReConnect = false;
	public int ConnectTimeOut = 10;
	@Override
	public void run() {
		if ((this.socketChannel == null) || (this.selector == null))
			return;
		isService = true;
		new Thread(new Runnable() {
			public void run() {
				MainProcess();
			}
		}, this.name + "-MainProcess").start();
		service();
	}

	public void MainProcess() {
		while (isService) {//isService
			try {
				if (getPacketData(this)) {
					OnCommand(this);
				}
				TimeUnit.MILLISECONDS.sleep(1);
			} catch (InterruptedException e) {
				System.out.println("取包服务停止interrupted");
				Thread.currentThread().interrupt();
			}
		}
	}

	public BaseCommandObject getFreeCommand() {
		try {
			return this.freeQueue.take();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		return null;
	}

	public boolean freeUsedCommand(BaseCommandObject command) {
		if (command == null)
			return false;
		try {
			return this.freeQueue.add(command);// 如果回收不了就抛异常，因为所有使用command都是free里取的
		} catch (Exception e) {
			Log.error("freeQueue:" + freeQueue.size(), e);
			return false;
		}
	}

	public abstract void OnCommand(Requestor requestor);

	public void close() {
		try {
			SelectionKey key = socketChannel.keyFor(selector);
			key.cancel();
			socketChannel.close();
		} catch (IOException e) {
			e.printStackTrace();
		}
	}

	public final ByteBuffer SendBuffer = ByteBuffer.allocate(1024 * 10);
	public final ByteBuffer RecvBuffer = ByteBuffer.allocate(1024 * 10);
	// 原子方式更新
	private final AtomicBoolean hasReceiveData = new AtomicBoolean(false);// 判断该session是否有接收的数据
	private final AtomicBoolean hasSendData = new AtomicBoolean(false);// 判断该session是否有发送的数据
	public BaseCommandObject dataPacket = new BaseCommandObject();
	private ArrayBlockingQueue<BaseCommandObject> freeQueue = null;
	public BaseCommandObject recDataPacket = new BaseCommandObject();
	public boolean isUseRecDataPacket = false;
	public boolean connected = false;
//	public StringBuilder StatusList = new StringBuilder();

	public String name;
	public String ip;
	public int port;

	public Requestor(String _name, String _ip, int _port) {
		super(_name + "-service");
		this.name = _name;
		this.ip = _ip;
		this.port = _port;
		SendBuffer.order(ByteOrder.LITTLE_ENDIAN);
		RecvBuffer.order(ByteOrder.LITTLE_ENDIAN);
		try {
			if (selector != null) {
				selector.close();
				selector = null;
			}
			selector = Selector.open();
		} catch (IOException e) {
			e.printStackTrace();
		}
		freeQueue = new ArrayBlockingQueue<BaseCommandObject>(GameData.SERVER.CmdThreadCount);
		for (int i = 0; i < GameData.SERVER.CmdThreadCount; i++) {
			freeQueue.add(new BaseCommandObject());
		}
	}

	public boolean connect() throws IOException {
		socketChannel = SocketChannel.open();
		InetSocketAddress isa = new InetSocketAddress(this.ip, this.port);
		socketChannel.connect(isa); // 采用阻塞模式连接服务器
		socketChannel.configureBlocking(false); // 设置为非阻塞模式
		this.register();
		connected = true;
		return true;
	}

	public void CMD_SNAPCARDS_RIVER() {
		this.readString();
		this.readEnd();
	}

	public void requestStandup() {
		synchronized (this.dataPacket) {
			this.setCmd(Protocol.EnterRoomCmd);
			this.writeBegin();
			this.writeByte((byte) 5);
			this.writeEnd();
			this.SendData();
		}
	}

	public boolean SendData(short cmdId, ByteBuffer bytes) {
		bytes.position(0);
		try {
			synchronized (this.SendBuffer) {

				if (SendBuffer.remaining() < bytes.remaining()) {
					Log.error(String.format("[Robot][%s]的发送缓冲已经不足,消息长度[%s]，请检查!", this, bytes.remaining()));
					return false;
				}
				SendBuffer.putInt(29099);
				SendBuffer.putInt(bytes.remaining() + 22);// length
				SendBuffer.putShort(cmdId);// cmd
				SendBuffer.putInt(1);// temp1
				SendBuffer.putInt(2);// temp2
				SendBuffer.putInt(3);// temp3
				SendBuffer.put(bytes);
				setHasSendData(true);
				send();
			}
		} catch (IOException e) {
			e.printStackTrace();
		}
		return true;
	}

	public void send() throws IOException {
//		if (!this.getHasSendData().get()) {
//			return;
//		}
//		synchronized (this.SendBuffer) {
		SendBuffer.flip(); // 把极限设为位置，把位置设为零
		while (this.getHasSendData().get()) {
			socketChannel.write(SendBuffer); // 发送数据，写的时候覆盖了之前的数据，
			// 一个完整包长度不够，设置没有数据
			if (SendBuffer.remaining() <= 0) {
				setHasSendData(false);
			}
			SendBuffer.compact(); // 删除已经发送的数据
		}
//		}
	}

	public boolean SendData(BaseCommandObject cmd) {
		try {
			return SendData(cmd.getCmd(), cmd.getBytes());
		} catch (Exception e) {
			return false;
		} finally {
			this.freeUsedCommand(cmd);
		}

	}

	public boolean SendData() {
		return SendData(dataPacket.getCmd(), dataPacket.getBytes());
	}

	public void register() {
		// 接收和发送数据
		try {
			selectorLock.lock();
			socketChannel.register(selector, SelectionKey.OP_READ, this);
		} catch (ClosedChannelException e1) {
			e1.printStackTrace();
		} finally {
			selectorLock.unlock();
		}
	}

	/**
	 * 提取一个包
	 * 
	 * @param sessionObj
	 * @return
	 */
	private boolean getPacketData(Requestor requestor) {
		// 相当于fireMessageReceived接受到，只是这里是读写线程之外的线程操作
		if (!requestor.getHasReceiveData().get()) {
			return false;
		}
		if (requestor.isUseRecDataPacket) {
			return false;
		}
		requestor.isUseRecDataPacket = true;
		boolean remain = false;
		ByteBuffer recvBuffer = requestor.RecvBuffer;
		synchronized (recvBuffer) {
			// 长度不够
			if (recvBuffer.position() < 22) {
				requestor.isUseRecDataPacket = false;
				return false;
			}
			int oldposition = recvBuffer.position();
			recvBuffer.flip();
			int remainLength = recvBuffer.remaining();
			int head = recvBuffer.getInt();
			if (head != 29099) {
				recvBuffer.position(0);
				recvBuffer.clear();// 本包出现问题，清空缓冲区，也可以断线处理
				printByteBufferInfo("[SessionMgr]本包出现问题 getPacketData>> RecvBuffer", recvBuffer);
				requestor.isUseRecDataPacket = false;
				throw new IllegalArgumentException("head not 29099");
//				return false;
			}
			int lenght = recvBuffer.getInt();
			// 粘包,头4个字节正确，只是包长度没有发完，只有继续等
			if (lenght > recvBuffer.remaining() + 8) {
				recvBuffer.position(oldposition);
				recvBuffer.limit(recvBuffer.capacity());
				printByteBufferInfo("[SessionMgr]包长度没有发完 getPacketData>> RecvBuffer", recvBuffer);
				requestor.isUseRecDataPacket = false;
				return false;
			}
			// printByteBufferInfo("<<getPacketData RecvBuffer", recvBuffer);
			// 提取数据操作
			BaseCommandObject dataPacket = requestor.recDataPacket;
			dataPacket.setCmd(recvBuffer.getShort());
//			System.out.println(dataPacket.getCmd());
			int dataLen = lenght - 22;
			// byte[] bytes = new byte[lenght - 22];
			// // 将数据部分提取
			// recvBuffer.get(bytes, 0, bytes.length);
			recvBuffer.position(22);
			while (dataLen > 0) {
				dataPacket.writeByte(recvBuffer.get());
				dataLen--;
			}
			// dataPacket.writeBegin();
//			dataPacket.putBytes(bytes);// 将包数据部分存入ByteBuffer
			dataPacket.writeEnd();
			// Log.debug(String.format("[RequestorMgr]接收到数据包,=====in==========in=========>命令:%s %s 总长度:%s 数据:%s",
			// dataPacket.cmdId, cmdSocket.get(dataPacket.cmdId), lenght,
			// dataPacket.byteToHex()));
			if (recvBuffer.remaining() > 0) {
				remain = true;
			}
			recvBuffer.compact();// 清除处理过的数据
			// 因为compact已经将
			// position恢复到limit,等待接收数据,limit到最大
			recvBuffer.position(remainLength - lenght);
			recvBuffer.limit(recvBuffer.capacity());
		}
		requestor.setHasReceiveData(remain);// 在下个取包操作时候，标志位,无包可以省去很多开销
		return true;
	}

	static final long SELECT_TIMEOUT = 1000L;
	int n;

	public void service() {
		Log.info("===========CenterRequestor-service开始侦听服务========");
		while (!Thread.currentThread().isInterrupted() && isService) {
			try {
				n = selector.select(SELECT_TIMEOUT);
				if (n > 0) {
					Set<SelectionKey> readyKeys = selector.selectedKeys();
					Iterator<SelectionKey> it = readyKeys.iterator();
					while (it.hasNext()) {
						SelectionKey key = null;
						try {
							key = it.next();
							it.remove();
							if (key.isReadable()) {
								receive(key);
							}
						} catch (IOException e) {
							e.printStackTrace();
							closeSessionKey(key);
						}
					}
				}
				TimeUnit.MILLISECONDS.sleep(1);

			} catch (IOException e1) {
				e1.printStackTrace();
				Log.error(String.format("[service]acceptor select() error%s ,break out", e1));
//				Thread.currentThread().interrupt();
				break;
			} catch (InterruptedException e) {
				e.printStackTrace();
				System.out.println("监听器服务停止interrupted");
//				Thread.currentThread().interrupt();
			}
		}
		Log.error("===========CenterRequestor-service停止服务========");
	}

	public void send(SelectionKey key) throws IOException {
		Requestor requestor = (Requestor) key.attachment();
		if (!requestor.getHasSendData().get()) {
			return;
		}
		synchronized (requestor.SendBuffer) {
			requestor.SendBuffer.flip(); // 把极限设为位置，把位置设为零
			requestor.socketChannel.write(requestor.SendBuffer); // 发送数据，写的时候覆盖了之前的数据，
			// 一个完整包长度不够，设置没有数据
			if (requestor.SendBuffer.remaining() <= 0) {
				requestor.setHasSendData(false);
			}
			requestor.SendBuffer.compact(); // 删除已经发送的数据
		}
	}

	private void printByteBufferInfo(String msg, ByteBuffer tempBuffer) {
		Log.info(String.format("%s remaining:%s position:%s limit:%s", msg, tempBuffer.remaining(), tempBuffer.position(), tempBuffer.limit()));
	}

	public boolean receive(SelectionKey key) throws IOException {
		Requestor requestor = (Requestor) key.attachment();
		SocketChannel socketChannel = (SocketChannel) key.channel();
		// 接收数据，可以从key中获得
		int num = 0;
		ByteBuffer tempBuffer = requestor.RecvBuffer;
		synchronized (tempBuffer) {
			num = socketChannel.read(tempBuffer);
			if (tempBuffer.position() > 0) {
				requestor.setHasReceiveData(true);
			}
		}
		if (num == -1) {
			closeSessionKey(key);
			return false;
		}
		return true;
	}

	private void closeSessionKey(SelectionKey key) {
		// 如果发生异常，取消该事件
		try {
			// 如果发生异常，取消该事件
			if (key != null) {
				Requestor requestor = (Requestor) key.attachment();
				requestor.connected = false;
				key.cancel();
			}
		} catch (Exception ex) {
			Log.info(ex);
		}
		onCloseSessionKey(key);

	}

	abstract void onCloseSessionKey(SelectionKey key);

	public AtomicBoolean getHasReceiveData() {
		return hasReceiveData;
	}

	public void setHasReceiveData(Boolean hasReceiveData) {
		this.hasReceiveData.set(hasReceiveData);
	}

	public AtomicBoolean getHasSendData() {
		return hasSendData;
	}

	public void setHasSendData(Boolean hasSendData) {
		this.hasSendData.set(hasSendData);
	}

	public ByteBuffer getBytes() {
		return dataPacket.getBytes();
	}

	public void setCmd(short cmd) {
		this.dataPacket.setCmd(cmd);
	}

	public short getCmd() {
		return this.dataPacket.getCmd();
	}

	public short getRecCmd() {
		return this.recDataPacket.getCmd();
	}

	public void setRecCmd(short cmd) {
		this.recDataPacket.setCmd(cmd);
	}

	// 写操作

	public void writeBegin() {
		dataPacket.writeBegin();
	}

	public void writeByte(byte value) {
		dataPacket.writeByte(value);
	}

	public void writeShort(Short value) {
		dataPacket.writeShort(value);
	}

	public void writeInt(int value) {
		dataPacket.writeInt(value);
	}

	public void writeFloat(float value) {
		dataPacket.writeFloat(value);
	}

	public void writeString(String msg) {
		dataPacket.writeString(msg);
	}

	public void writeEnd() {
		dataPacket.flip();
	}

	// 读取操作
	public void readEnd() {
//		recDataPacket.dataPacket.byteBuffer.compact();
		recDataPacket.clear();
		this.isUseRecDataPacket = false;
	}

	public char readChar() {
		return recDataPacket.readChar();
	}

	public byte readByte() {
		return recDataPacket.readByte();
	}

	public float readFloat() {
		return recDataPacket.readFloat();
	}

	public long readLong() {
		return recDataPacket.readLong();
	}

	public double readDouble() {
		return recDataPacket.readDouble();
	}

	public int readInt() {
		return recDataPacket.readInt();
	}

	public short readShort() {
		return recDataPacket.readShort();
	}

	public String readString() {
		return recDataPacket.readString();
	}

	public ByteBuffer pack() {
		int l = length();
		ByteBuffer byteBufferer = ByteBuffer.allocate(l);
		if (position() > 0) {
			flip();
		}
		byteBufferer.put(array(), 0, l);
		byteBufferer.flip();
		return byteBufferer;
	}

	public byte[] array() {
		return dataPacket.array();
	}

	public int position() {
		return dataPacket.position();
	}

	public void position(int newPosition) {
		dataPacket.position(newPosition);
	}

	public int remaining() {
		return dataPacket.remaining();
	}

	public int limit() {
		return dataPacket.limit();
	}

	public void flip() {
		dataPacket.flip();
	}

	public void clear() {
		dataPacket.clear();
	}

	public int length() {
		return dataPacket.length();
	}

	public int totalSize() {
		return dataPacket.totalSize();
	}

	public void outInfo(byte[] bytes) {
		dataPacket.outInfo(bytes);
	}

}
